package com.fanxl.flink.basic;

import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/1/11 14:06
 */
public class StreamingControllerApp {

    private static Logger log = LoggerFactory.getLogger(StreamingControllerApp.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream<Integer> configStream = env.addSource(new SourceFunction<Integer>() {

            private boolean running = true;

            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                sourceContext.collect(2);
            }

            @Override
            public void cancel() {
                running = false;
            }
        }).broadcast();

        DataStream<Integer> dataStream = env.addSource(new SourceFunction<Integer>() {

            private boolean running = true;

            @Override
            public void run(SourceContext<Integer> sourceContext) throws Exception {
                for (int i = 0; running && i < 20; i += 1) {
                    sourceContext.collect(i);
                }
            }

            @Override
            public void cancel() {
                running = false;
            }
        });

        ConnectedStreams connectedStreams = dataStream.connect(configStream);
        DataStream<Integer> resultStream = connectedStreams.process(new CoProcessFunction<Integer, Integer, Integer>(){

            private Integer divisor = 1;

            @Override
            public void processElement1(Integer o, Context context, Collector<Integer> collector) throws Exception {
                log.info("[Data] Received {}, divisor {}", o, divisor);
                if (o % divisor == 0) {
                    collector.collect(o);
                }
            }

            @Override
            public void processElement2(Integer o, Context context, Collector<Integer> collector) throws Exception {
                log.info("[Divisor] Received {}", o);
                divisor = o;
            }
        });
        resultStream.print();
        env.execute();

    }

}
